package com.leo.hbase.manager.web.hds;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.github.CCweixiao.hbase.sdk.common.constants.HBaseConfigKeys;
import com.github.CCweixiao.hbase.sdk.common.exception.HBaseOperationsException;
import com.github.CCweixiao.hbase.sdk.common.exception.HBaseSdkConnectionException;
import com.github.CCweixiao.hbase.sdk.common.util.StringUtil;
import com.github.CCweixiao.hbase.sdk.shell.HBaseShellSession;
import com.github.CCweixiao.hbase.sdk.shell.HBaseShellSessionManager;
import com.github.CCweixiao.hbase.sdk.template.*;
import com.leo.hbase.manager.common.exception.BusinessException;
import com.leo.hbase.manager.common.utils.SocketUtil;
import com.leo.hbase.manager.system.domain.SysHbaseCluster;
import com.leo.hbase.manager.system.dto.PropertyDto;
import com.leo.hbase.manager.system.dto.SysHbaseClusterDto;
import com.leo.hbase.manager.system.service.ISysHbaseClusterService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.Properties;

/**
 * HBase动态数据源管理器
 *
 * @author leojie 2020/9/17 10:34 下午
 */
@Component
public class HBaseClusterDSConfig {
    private static final Logger LOG = LoggerFactory.getLogger(HBaseClusterDSConfig.class);

    @Autowired
    private ISysHbaseClusterService hbaseClusterService;

    public BaseHBaseTableTemplate getHBaseTemplate(String clusterCode) {
        LOG.info("当前获取集群:{}的HBaseTemplate对象！", clusterCode);
        Properties properties = getHBaseProperties(clusterCode);
        return HBaseTableTemplate.of(properties);
    }

    public BaseHBaseSqlTemplate getHBaseSqlTemplate(String clusterCode){
        LOG.info("当前获取集群:{}的HBaseSqlTemplate对象！", clusterCode);
        Properties properties = getHBaseProperties(clusterCode);
        return HBaseSqlTemplate.of(properties);
    }

    public BaseHBaseAdminTemplate getHBaseAdminTemplate(String clusterCode) {
        LOG.info("当前获取集群:{}的HBaseAdminTemplate对象！", clusterCode);
        Properties properties = getHBaseProperties(clusterCode);
        return HBaseAdminTemplate.of(properties);
    }

    public BaseHBaseAdminTemplate getHBaseAdminTemplate(SysHbaseClusterDto sysHbaseClusterDto) {
        String clusterId = sysHbaseClusterDto.getClusterId();
        LOG.info("当前获取集群:{}的HBaseAdminTemplate对象！", clusterId);
        Properties properties = getHBaseProperties(sysHbaseClusterDto);
        return HBaseAdminTemplate.of(properties);
    }

    public HBaseShellSession getHBaseShellSession(String clusterCode) {
        LOG.info("当前获取集群:{}的HBaseShellSession对象！", clusterCode);
        Properties properties = getHBaseProperties(clusterCode);
        return HBaseShellSessionManager.getHBaseShellSession(properties);
    }

    private Properties getHBaseProperties(String clusterId) {
        LOG.info("开始解析HBase集群:{}的配置......", clusterId);
        SysHbaseCluster sysHbaseCluster = hbaseClusterService.selectSysHbaseClusterById(clusterId);
        if (sysHbaseCluster == null || sysHbaseCluster.getId() < 0) {
            throw new HBaseOperationsException(String.format("The cluster %s is not exists.", clusterId));
        }
        SysHbaseClusterDto sysHbaseClusterDto = new SysHbaseClusterDto().convertFor(sysHbaseCluster);
        return getHBaseProperties(sysHbaseClusterDto);
    }

    private Properties getHBaseProperties(SysHbaseClusterDto sysHbaseClusterDto) {
        SysHbaseCluster sysHbaseCluster = sysHbaseClusterDto.convertTo();
        String clusterConfig = sysHbaseCluster.getClusterConfig();
        if (StringUtil.isNotBlank(clusterConfig) && !"null".equals(clusterConfig)) {
            JSONArray propArr = JSON.parseArray(clusterConfig);
            Properties properties = new Properties();
            for (int i = 0; i < propArr.size(); i++) {
                PropertyDto propObj = propArr.getObject(i, PropertyDto.class);
                properties.setProperty(propObj.getPropertyName(), propObj.getPropertyValue());
            }
            String zkHost = properties.getProperty(HBaseConfigKeys.ZOOKEEPER_QUORUM);
            if (StringUtil.isBlank(zkHost)) {
                zkHost = "localhost";
            }
            String zkPort = properties.getProperty(HBaseConfigKeys.ZOOKEEPER_CLIENT_PORT);
            if (StringUtil.isNotBlank(zkPort)) {
                zkPort = "2181";
            }
            String[] zkHostArr = zkHost.split(",");
            for (String zk : zkHostArr) {
                boolean telnet = SocketUtil.telnet(zk, Integer.parseInt(zkPort), 10 * 1000);
                if (!telnet) {
                    throw new HBaseSdkConnectionException(String.format("zkHost:[%s],zkPort:[%s]访问不通", zk, zkPort));
                }
            }

            return properties;
        } else {
            throw new BusinessException("解析HBase集群配置异常");
        }
    }
}
